package com.ibm.cps.sql;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import net.sf.jsqlparser.expression.Expression;
import net.sf.jsqlparser.schema.Table;
import net.sf.jsqlparser.statement.select.FromItem;
import net.sf.jsqlparser.statement.select.PlainSelect;
import net.sf.jsqlparser.statement.select.SelectItem;
import net.sf.jsqlparser.statement.select.SelectVisitorAdapter;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.sql.expression.BooleanExpression;
import com.ibm.cps.sql.expression.ConditionCompositionExpression;
import com.ibm.cps.sql.expression.ConditionCompositionExpression.Predicate;
import com.ibm.cps.sql.netsfparser.ConditionParser;
import com.ibm.cps.sql.netsfparser.FromItemParser;
import com.ibm.cps.sql.netsfparser.GroupByKeyParser;
import com.ibm.cps.sql.window.WindowDescriptor;
import com.ibm.cps.sql.window.WindowingRequestParser;
import com.ibm.factories.DataSourceFactory;
import com.ibm.util.ErrorCode;
import com.ibm.util.SchemaCache;
import com.ibm.util.exception.CPSException;

public class SelectBodyVisitor extends SelectVisitorAdapter {

	private String ruleName;
	private String groupId;
	private JsonNode sqlMeta;
	private ArrayNode resultingMetaData;
	private String tenantid;
	private DataSourceFactory datasourceFactory;
	private SchemaCache schemaCache;

	public SelectBodyVisitor(ObjectNode sqlMeta, String tenantid, DataSourceFactory datasourceFactory, SchemaCache schemaCache) {
		this.ruleName = sqlMeta.get(MessageFields.NAME).asText();
		this.groupId = sqlMeta.get(MessageFields.GROUPID).asText();
		this.sqlMeta = sqlMeta;
		this.tenantid = tenantid;
		this.datasourceFactory = datasourceFactory;
		this.schemaCache = schemaCache;
	}

	@Override
	public void visit(PlainSelect plainSelect) {
		try {
			// select items, alias map to its Expression
			SelectItemDescriptor outputFields = parseSelectItems(plainSelect);
			// from xxx
			List<String> dataSources = getDataSources(plainSelect);
			// into xxx
			List<String> publishers = getPublishers(plainSelect);
			// groupby xxx
			AggregationDescriptor aggregation = parseAggregation(plainSelect);
			// Where and having
			BooleanExpression filteringCondition = getFilteringCondition(plainSelect);

			if (aggregation.getWindow() == null) {
				NonWindowingRequestParser parser = new NonWindowingRequestParser(datasourceFactory, schemaCache, ruleName, groupId, tenantid);
				ArrayNode metaData = parser.parseNonWindowingRequest(dataSources, outputFields, aggregation, filteringCondition, publishers);
				metaData.add(sqlMeta);
				resultingMetaData = metaData;
			} else {
				WindowingRequestParser parser = new WindowingRequestParser(datasourceFactory, schemaCache, ruleName, groupId, tenantid);
				ArrayNode metaData = parser.parseWindowingRequest(dataSources, outputFields, aggregation, filteringCondition, publishers);
				metaData.add(sqlMeta);
				resultingMetaData = metaData;
			}
		} catch (CPSException ex) {
			throw new IllegalArgumentException(ex);
		}
	}
	
	private SelectItemDescriptor parseSelectItems(PlainSelect select) throws CPSException{
		SelectItemParser outputFieldParser = new SelectItemParser();
		List<SelectItem> items = select.getSelectItems();
		if(items == null){
			throw new CPSException(ErrorCode.INPUT_ERROR, "No field is selected in this stream query");
		}
		SelectItemDescriptor projection = outputFieldParser.parse(items);
		return projection;
	}

	private List<String> getDataSources(PlainSelect select) throws CPSException{
		FromItem fromItem = select.getFromItem();
		if (fromItem != null) {
			FromItemParser fromItemParser = new FromItemParser();
			fromItem.accept(fromItemParser);
			List<String> dataSources = fromItemParser.getDataSources();
			//Join grammar is not supported by now, should it be on timestamp only?
			return dataSources;
		}else {
			throw new CPSException(ErrorCode.INPUT_ERROR, "Data source is not specified. Please use into <datasource name> clause to specify data source");
		}
	}

	private List<String> getPublishers(PlainSelect select) {
		List<Table> outputStream = select.getIntoTables();
		List<String> outputNames = new ArrayList<>();
		if (outputStream == null) {
//			throw new IllegalArgumentException("Please specify output stream using INTO <Output Name> clause");
			// modify by gxj, return a default output when publishers are not specified
			outputNames.add(MessageFields.UNDEFINED_PROCESSOR);
			return outputNames;
		}
		for (Table t : outputStream) {
			outputNames.add(t.getFullyQualifiedName());
		}
		return outputNames;
	}
	
	private AggregationDescriptor parseAggregation(PlainSelect select) throws CPSException{
		List<Expression> groupKeys = select.getGroupByColumnReferences();
		WindowDescriptor timeWindow = null;
		Set<String> aggregationKeys = new HashSet<>();
		if(groupKeys != null){
			for (Expression expression : groupKeys) {
				GroupByKeyParser parser = new GroupByKeyParser();
				try{
					expression.accept(parser);
				}catch(IllegalArgumentException parseError){
					throw new CPSException(ErrorCode.INPUT_ERROR, parseError.getMessage());
				}
				if(parser.getAggregationField() != null){
					aggregationKeys.add(parser.getAggregationField());
				}else{
					WindowDescriptor window = parser.getWindow();
					if(timeWindow != null)
						throw new CPSException(ErrorCode.INPUT_ERROR, "Only one window declaration allowed in each stream query");
					timeWindow = window;
				}
			}
		}
		if(!aggregationKeys.isEmpty() && timeWindow != null)
			throw new CPSException(ErrorCode.INPUT_ERROR, "Window declaration is required to perform aggregation on fields: " + aggregationKeys);
		return new AggregationDescriptor(timeWindow, aggregationKeys);
	}
	
	private BooleanExpression getFilteringCondition(PlainSelect select){
		ConditionParser parser = new ConditionParser();
		BooleanExpression whereCondition = null;
		BooleanExpression havingCondition = null;
		Expression whereExpression = select.getWhere();
		if(whereExpression != null){
			whereExpression.accept(parser);
			whereCondition = parser.getExpression();
		}
		Expression havingExpression = select.getHaving();
		if(havingExpression != null){
			havingExpression.accept(parser);
			havingCondition = parser.getExpression();
		}
		if(whereCondition == null)
			return havingCondition;
		if(havingCondition == null)
			return whereCondition;
		return new ConditionCompositionExpression(whereCondition, havingCondition, Predicate.AND);
	}

	public ArrayNode getResult() {
		return resultingMetaData;
	}
}
